package com.codrim.cs.data.server.rpc;

import com.alibaba.dubbo.config.annotation.Service;
import com.codrim.common.utils.enums.EnumWithKey;
import com.codrim.common.utils.exception.Assert;
import com.codrim.common.utils.json.JsonMapper;
import com.codrim.common.utils.web.vo.PagingResult;
import com.codrim.cs.data.rpc.OfferReportService;
import com.codrim.cs.data.rpc.dto.param.AdvertiserReportParamDTO;
import com.codrim.cs.data.rpc.dto.param.AffiliateReportParamDTO;
import com.codrim.cs.data.rpc.dto.param.BillInOutReportParam;
import com.codrim.cs.data.rpc.dto.param.OfferReportField;
import com.codrim.cs.data.rpc.dto.param.PageResult;
import com.codrim.cs.data.rpc.dto.param.ReportBaseParamDTO;
import com.codrim.cs.data.rpc.dto.param.TopOfferConversionParamDTO;
import com.codrim.cs.data.rpc.dto.result.BillInOutReportDTO;
import com.codrim.cs.data.rpc.dto.result.OfferReportDataDTO;
import com.codrim.cs.data.rpc.dto.result.PagingResultWithSumData;
import com.codrim.cs.data.rpc.dto.result.TopOfferConversionResultDTO;
import com.codrim.cs.data.rpc.enums.OfferReportType;
import com.codrim.cs.data.rpc.enums.TimeDimension;
import com.codrim.cs.data.server.dao.RealTimeDataDao;
import com.codrim.cs.data.server.dao.entity.MergeDataRecordDO;
import com.codrim.cs.data.server.dao.entity.MergeDayStatCsGeneralDO;
import com.codrim.cs.data.server.dao.entity.TopOfferConversionDO;
import com.codrim.cs.data.server.dao.mapper.AffiliateMergeColumnAndAlias;
import com.codrim.cs.data.server.dao.mapper.DataRecordMapper;
import com.codrim.cs.data.server.dao.mapper.StatCsAdvOfferMapper;
import com.codrim.cs.data.server.dao.mapper.StatCsGeneralMapper;
import com.codrim.cs.data.server.dao.param.OrderByColumn;
import com.codrim.cs.data.server.dao.param.RealTimeDataSearchParam;
import com.codrim.cs.data.server.dao.param.ReportParam;
import com.codrim.cs.data.server.dao.param.SelectColumnNameAndAlias;
import com.codrim.cs.data.server.dao.realtime.AllRedisRealTimeData;
import com.codrim.cs.data.server.dao.realtime.RealTimeData;
import com.codrim.cs.data.server.dao.realtime.RealTimeGroupByField;
import com.codrim.cs.data.server.subservice.TimeZoneSubService;
import com.codrim.cs.data.server.utils.DateUtils;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.assertj.core.util.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoField;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalAccessor;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import javax.annotation.Resource;
import javax.validation.Valid;

import lombok.Data;

import static com.codrim.cs.data.rpc.dto.param.AdvertiserReportParamDTO.AdvertiserOfferReportGroupByField;
import static com.codrim.cs.data.rpc.dto.param.AffiliateReportParamDTO.AffiliateOfferReportGroupByField;
import static com.codrim.cs.data.server.subservice.ReportUtils.timeDimensionWithSdate;

/**
 * @author liang.ma
 */
@Service(version = OfferReportService.RPC_VERSION)
public class OfferReportServiceImpl implements OfferReportService {

    private static final Logger logger = LoggerFactory.getLogger(OfferReportServiceImpl.class);

    @Resource private StatCsGeneralMapper dayStatCsGeneralMapper;
    @Resource private TimeZoneSubService timeZoneSubService;
    @Resource private DataRecordMapper dataRecordMapper;
    @Resource private RealTimeDataDao realTimeDataDao;
    @Resource private StatCsAdvOfferMapper dayStatCsAdvOfferMapper;

    @Override
    public PagingResultWithSumData<OfferReportDataDTO> offerReportForAdvert(@Valid AdvertiserReportParamDTO param) {
        logger.info("param:{}", JsonMapper.nonEmptyMapper().toJson(param));

        final TimeDimension originTimeDimension = param.getTimeDimension();

        param.setTimeDimension(TimeDimension.Day);

        final List<OfferReportField> options = param.getOfferReportFieldList();
        Assert.notNull(options, "Options not exists");
        ReportParam reportParam = validAndConstructReportParam(param);
        reportParam.setDataOwner(ReportParam.DataOwner.ADVERTISER);
        reportParam.setAdvAdvertiserManagerIdSelected(param.getAdvertiserManagerIdSelected());
        reportParam.setAdvAffiliateManagerIdSelected(param.getAffiliateManagerSelected());
        reportParam.setAdvAdvertiserManagerDepartmentIdListSelected(param.getAdvAdvertiserManagerDepartmentIdListSelected());
        reportParam.setAdvAffiliateManagerDepartmentIdListSelected(param.getAdvAffiliateManagerDepartmentIdListSelected());
        reportParam.setAdvAdvertiserManagerIdList(param.getAdvAdvertiserManagerIdList());
        reportParam.setAdvAffiliateManagerIdList(param.getAdvAffiliateManagerIdList());
        reportParam.setCustomerIdList(param.getAdvertiserIdList());
        reportParam.setAffIds(Optional.ofNullable(param.getAffiliateIds())
                .map(item -> item.stream().map(String::valueOf).collect(Collectors.toList()))
                .orElse(null));

        Optional.ofNullable(param.getAccessType())
                .map(EnumWithKey::getKey)
                .ifPresent(reportParam::setAccessType);

        TimeDimension timeDimension = param.getTimeDimension();
        final Integer timeZone = param.getTimeZone();
        final String zone = timeZoneSubService.prefixTableNameForTimeZone(timeZone);

        final ReportDataHelper pageDataHelper = new ReportDataHelper();
        final ReportDataHelper sumDataHelper = new ReportDataHelper();
        //该记录为了统计实时数据里面按人民币汇总的数据
        MergeDayStatCsGeneralDO mergeRealTimeByRMB = MergeDayStatCsGeneralDO.constructBySymbolRMB();
        //该记录为了统计实时数据里面按美元汇总的数据
        MergeDayStatCsGeneralDO mergeRealTimeByDollar = MergeDayStatCsGeneralDO.constructBySymbolDoller();

        try {
             /* 1. 检索自动生成的报表数据 */
            List<MergeDayStatCsGeneralDO> sumStatGeneral = null;
            int totalCsGeneral = 0;
            if (Optional.ofNullable(param.getRequireTracking()).orElse(true)) {
                final List<SelectColumnNameAndAlias> csGeneralSelectColumns = options.stream()
                        .map(StatCsGeneralMapper.AdvertiserMergeColumnAndAlias::select)
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList());
                handleCsGeneralDataSelectColumnSdate(StatCsGeneralMapper.AdvertiserMergeColumnAndAlias.Sdate.getSelectColumnNameAndAlias(),
                        csGeneralSelectColumns, originTimeDimension);
                totalCsGeneral = dayStatCsGeneralMapper.countDayStatCsGeneral(reportParam,
                        options.stream().map(StatCsGeneralMapper.AdvertiserMergeColumnAndAlias::groupBy)
                                .filter(Objects::nonNull).collect(Collectors.toList()), zone, timeDimension.key);

                if (totalCsGeneral > 0) {
                    sumStatGeneral = dayStatCsGeneralMapper.sumDayStatCsGeneral(reportParam, zone, timeDimension.key);
                    assembleAdvertiserCsGeneralData(pageDataHelper, param, reportParam,
                            mergeRealTimeByRMB, mergeRealTimeByDollar,
                            originTimeDimension, csGeneralSelectColumns);
                }
                logger.info("csGeneral, mapDataSize:{}", totalCsGeneral);
            }

             /* 2. 检索手动录入数据 */
            int totalDataRecord = 0;
            List<MergeDataRecordDO> sumDataRecord = null;
            Boolean requireTracking = param.getRequireTracking();
            if (timeDimension != TimeDimension.Hour && reportParam.getReportType() == OfferReportType.NORMAL_REPORT
                    && Optional.ofNullable(requireTracking).map(tmp -> !tmp).orElse(true)) {
                    /* 手动录入数据 仅为天数据  故需要转换时间查询区间(可能为年,月,周...)到天区间 */
                reportParam.setStartDate(timeDimensionWithSdate(param.getStartDate(), timeZone, TimeDimension.Day).getSdate());
                reportParam.setEndDate(timeDimensionWithSdate(param.getEndDate(), timeZone, TimeDimension.Day).getSdate());

                final List<String> dataRecordGroupByColumns = options.stream()
                        .map(DataRecordMapper.AdvertiserMergeColumnAndAlias::groupBy)
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList());

                final SelectColumnNameAndAlias sdateColumnNameAndAlias = DataRecordMapper.AdvertiserMergeColumnAndAlias.Sdate.getSelectColumnNameAndAlias();

                handleDataRecordGroupByColumnSdate(sdateColumnNameAndAlias,
                        dataRecordGroupByColumns, timeDimension);

                totalDataRecord = dataRecordMapper.countDataRecord(reportParam, dataRecordGroupByColumns, zone);
                logger.info("dataRecord, mapDataSize:{}", totalDataRecord);
                sumDataRecord = dataRecordMapper.sumDataRecord(reportParam, zone);
                if (reportParam.getRows() > pageDataHelper.data.size() && totalDataRecord > 0) {
                    /* 最后一条数据所处的位置 不超过totalDataRecord的第一页 */
                    updatePageParam(reportParam, totalCsGeneral, pageDataHelper.data.size());
                    final List<SelectColumnNameAndAlias> dataRecordSelectColumns = options.stream()
                            .map(DataRecordMapper.AdvertiserMergeColumnAndAlias::select)
                            .filter(Objects::nonNull)
                            .collect(Collectors.toList());
                    handelDataRecordSelectColumnSdate(sdateColumnNameAndAlias,
                            dataRecordSelectColumns, originTimeDimension);
                    List<MergeDataRecordDO> dataRecordDataList = dataRecordMapper.mergeDataRecord(
                            reportParam, dataRecordSelectColumns, dataRecordGroupByColumns, zone);
                    pageDataHelper.addAllMergeDataRecord(dataRecordDataList);
                }
            }

            /* 3 处理合计数据 */
            if (sumStatGeneral != null) {
                sumStatGeneral.forEach(sumDataHelper::add);
            }

            if (sumDataRecord != null) {
                sumDataRecord.forEach(sumDataHelper::add);
            }
            sumDataHelper.add(mergeRealTimeByRMB);
            sumDataHelper.add(mergeRealTimeByDollar);

            return new PagingResultWithSumData<>(
                    new PagingResult<>(totalCsGeneral + totalDataRecord, pageDataHelper.data()), sumDataHelper.data());
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return null;
    }

    private void assembleAdvertiserCsGeneralData(ReportDataHelper helper, AdvertiserReportParamDTO param,
            ReportParam reportParam, MergeDayStatCsGeneralDO mergeRealTimeByRMB,
            MergeDayStatCsGeneralDO mergeRealTimeByDollar, TimeDimension originTimeDimension,
            List<SelectColumnNameAndAlias> csGeneralSelectColumns) {

        final List<OfferReportField> options = param.getOfferReportFieldList();

         /* 1. 检索自动生成的报表数据 */
        final List<String> csGeneralGroupByColumns = options.stream()
                .map(StatCsGeneralMapper.AdvertiserMergeColumnAndAlias::groupBy)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        List<OrderByColumn> csGeneralOrderByColumns = options.stream()
                .map(StatCsGeneralMapper.AdvertiserMergeColumnAndAlias::orderBy)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(csGeneralOrderByColumns)) {
            csGeneralOrderByColumns = new ArrayList<>();
        }
        csGeneralOrderByColumns.add(new OrderByColumn("id", "desc"));

        final List<MergeDayStatCsGeneralDO> statCsGeneralDataList = dayStatCsGeneralMapper.mergeDataFromDayStatCsGeneral(
                reportParam, csGeneralSelectColumns, csGeneralGroupByColumns,
                csGeneralOrderByColumns, timeZoneSubService.prefixTableNameForTimeZone(param.getTimeZone()),
                TimeDimension.Day.key);
        JsonMapper jsonMapper = JsonMapper.nonEmptyMapper();
        logger.info("csGeneral reportParam:{},selectParam:{}, groupByColumns:{}, orderByColumns:{} mapDataSize:{}",
                jsonMapper.toJson(reportParam), jsonMapper.toJson(csGeneralSelectColumns),
                jsonMapper.toJson(csGeneralGroupByColumns), jsonMapper.toJson(csGeneralOrderByColumns),
                statCsGeneralDataList.size());
        helper.addAllDayStatCsGeneralMergeData(statCsGeneralDataList);

        /* 2. 添加实时数据 */
        /* 仅支持 天报表 和小时报表的实时数据  */
        boolean hourOrDay = originTimeDimension == TimeDimension.Hour || originTimeDimension == TimeDimension.Day;
        boolean includeNow = isIncludeNow(param);
        boolean advertiserGroupMath = advertiserGroupMatch(options);
        logger.info("inCloudeNow:{}, affilateGroupMatch:{}", includeNow, advertiserGroupMath);
        if (param.getReportType() != OfferReportType.AlternativeReport && hourOrDay && includeNow && advertiserGroupMath) {
            /**
             * TODO: 优化实时数据的key,区分数据属于广告报表或者备选报表
             *  备选报表不支持实时数据,原因: 实时数据暂时无法区分是备选报表还是广告报表的数据,和诸葛商量,暂时把实时数据都默认当做广告报表的数据
             */
            List<RealTimeGroupByField> collect = options.stream()
                    .map(RealTimeGroupByField::of)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
            addRealTimeData(param.getTimeZone(), param.getTimeDimension(),
                    collect, helper, mergeRealTimeByRMB, mergeRealTimeByDollar);
        }
    }

    private void updatePageParam(ReportParam reportParam, int totalCsGeneral, int csGeneralSizeThisTime) {
         /* 最后一条数据所处的位置 不超过totalDataRecord的第一页 */
        if (reportParam.getFromLine() + reportParam.getRows() - totalCsGeneral <= reportParam.getRows()) {
            reportParam.setFromLine(0);
            reportParam.setRows(reportParam.getRows() - csGeneralSizeThisTime);
        } else {
            reportParam.setFromLine(reportParam.getFromLine() - totalCsGeneral);
            reportParam.setRows(reportParam.getRows());
        }
    }

    @SuppressWarnings("all")
    private boolean advertiserGroupMatch(List<OfferReportField> options) {
        for (OfferReportField offerReportField : options) {
            if (ADVERTISER_NOT_MATCH_GROUPS.contains(offerReportField.getField())) {
                return false;
            }
        }
        return true;
    }

    @Override
    public PagingResultWithSumData<OfferReportDataDTO> offerReportForAffiliate(
            @Valid AffiliateReportParamDTO param) {
        logger.info("param:{}", JsonMapper.getInstance().toJson(param));

        final TimeDimension originTimeDimension = param.getTimeDimension();
        param.setTimeDimension(TimeDimension.Day);

        final List<OfferReportField> options = param.getOfferReportFieldList();
        Assert.notNull(options, "Options not exists");
        ReportParam reportParam = validAndConstructReportParam(param);
        reportParam.setDataOwner(ReportParam.DataOwner.AFFILIATE);
        reportParam.setAdvIds(Optional.ofNullable(param.getAdvertiserIds())
                .map(item -> item.stream().map(String::valueOf).collect(Collectors.toList())).orElse(null));
        reportParam.setAffAdvertiserManagerIdList(param.getAffAdvertiserManagerIdList());
        //部门
        reportParam.setAffDepartmentIds(param.getAffDepartmentIds());
        //商务经理
        reportParam.setAffAdvertiserManagerId(param.getAffAdvertiserManagerId());
        TimeDimension timeDimension = param.getTimeDimension();
        final String zone = timeZoneSubService.prefixTableNameForTimeZone(param.getTimeZone());
        final ReportDataHelper pageDataHelper = new ReportDataHelper();
        final ReportDataHelper sumDataHelper = new ReportDataHelper();
        //该记录为了统计实时数据里面按人民币汇总的数据
        MergeDayStatCsGeneralDO mergeRealTimeByRMB = MergeDayStatCsGeneralDO.constructBySymbolRMB();
        //该记录为了统计实时数据里面按美元汇总的数据
        MergeDayStatCsGeneralDO mergeRealTimeByDollar = MergeDayStatCsGeneralDO.constructBySymbolDoller();

            /* 1. 检索自动生成的报表数据 */
        int totalCsGeneral = 0;
        List<MergeDayStatCsGeneralDO> sumStatGeneral = null;
        if (Optional.ofNullable(param.getRequireTracking()).orElse(true)) {
            List<String> collect = options.stream()
                    .map(AffiliateMergeColumnAndAlias::groupBy)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());

            final List<SelectColumnNameAndAlias> csGeneralSelectColumns = options.stream()
                    .map(AffiliateMergeColumnAndAlias::select)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());

            handleCsGeneralDataSelectColumnSdate(
                    AffiliateMergeColumnAndAlias.Sdate.getSelectColumnNameAndAlias(),
                    csGeneralSelectColumns, originTimeDimension);
            totalCsGeneral = dayStatCsGeneralMapper.countDayStatCsGeneral(reportParam,
                    collect, zone, timeDimension.key);

            if (totalCsGeneral > 0) {
                sumStatGeneral = dayStatCsGeneralMapper.sumDayStatCsGeneral(reportParam, zone, timeDimension.key);
                assembleAffiliateCsGeneralData(pageDataHelper, param, reportParam,
                        timeDimension, mergeRealTimeByRMB, mergeRealTimeByDollar, originTimeDimension,
                        csGeneralSelectColumns);
            }
            logger.info("csGeneral, mapDataSize:{}", totalCsGeneral);
        }

        /* 2. 检索手动录入数据 */
        int totalDataRecord = 0;
        List<MergeDataRecordDO> sumDataRecord = null;
        if (timeDimension != TimeDimension.Hour && reportParam.getReportType() == OfferReportType.NORMAL_REPORT
                && Optional.ofNullable(param.getRequireTracking()).map(tmp -> !tmp).orElse(true)) {
                  /* 手动录入数据 仅为天数据  故需要转换时间查询区间(可能为年,月,周...)到天区间 */
            reportParam.setStartDate(timeDimensionWithSdate(param.getStartDate(), param.getTimeZone(), TimeDimension.Day).getSdate());
            reportParam.setEndDate(timeDimensionWithSdate(param.getEndDate(), param.getTimeZone(), TimeDimension.Day).getSdate());

            final List<String> dataRecordGroupByColumns = options.stream()
                    .map(DataRecordMapper.AffiliateMergeColumnAndAlias::groupBy)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
            SelectColumnNameAndAlias sDateColumnAndAlias = DataRecordMapper.AffiliateMergeColumnAndAlias.Sdate.getSelectColumnNameAndAlias();
            handleDataRecordGroupByColumnSdate(sDateColumnAndAlias, dataRecordGroupByColumns, timeDimension);
            totalDataRecord = dataRecordMapper.countDataRecord(reportParam, dataRecordGroupByColumns, zone);
            logger.info("dataRecord, mapDataSize:{}", totalDataRecord);
            sumDataRecord = dataRecordMapper.sumDataRecord(reportParam, zone);
            /* reportParam.getRows() > csGeneralSizeThisTime 为true时, 则已查到的数据 不足一页,需要从dataRecord中取数据补足一页 */
            if (reportParam.getRows() > pageDataHelper.data.size() && totalDataRecord > 0) {
                updatePageParam(reportParam, totalCsGeneral, pageDataHelper.data.size());
                List<SelectColumnNameAndAlias> dataRecordSelectColumns = options.stream()
                        .map(DataRecordMapper.AffiliateMergeColumnAndAlias::select)
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList());
                handelDataRecordSelectColumnSdate(sDateColumnAndAlias, dataRecordSelectColumns, timeDimension);
                List<MergeDataRecordDO> dataRecordList = dataRecordMapper.mergeDataRecord(
                        reportParam, dataRecordSelectColumns, dataRecordGroupByColumns, zone);
                pageDataHelper.addAllMergeDataRecord(dataRecordList);
            }
        }

        /* 3 处理合计数据 */
        if (sumStatGeneral != null) {
            sumStatGeneral.forEach(sumDataHelper::add);
        }
        if (sumDataRecord != null) {
            sumDataRecord.forEach(sumDataHelper::add);
        }
        sumDataHelper.add(mergeRealTimeByRMB);
        sumDataHelper.add(mergeRealTimeByDollar);
        PagingResultWithSumData<OfferReportDataDTO> result = new PagingResultWithSumData<>(
                new PagingResult<>(totalCsGeneral + totalDataRecord, pageDataHelper.data()), sumDataHelper.data());
        logger.debug("Result:{}", result);
        return result;
    }

    /**
     * 根据时间纬度,调整DataRecord.sdate的查询字段格式化
     *
     * @param target                  目标字段
     * @param dataRecordSelectColumns 显示字段
     * @param timeDimension           时间纬度
     */
    private void handelDataRecordSelectColumnSdate(SelectColumnNameAndAlias target,
            List<SelectColumnNameAndAlias> dataRecordSelectColumns, TimeDimension timeDimension) {
        if (CollectionUtils.isNotEmpty(dataRecordSelectColumns)) {
            dataRecordSelectColumns.forEach(item -> {
                if (item.columnName.equals(target.columnName)) {
                    item.setColumnName(dataRecordColumnNameOfSdate(timeDimension));
                }
            });
        }
    }

    /**
     * 根据时间纬度,调整DataRecord.sdate的分组字段格式化
     *
     * @param dataRecordGroupColumns 分组字段
     * @param timeDimension          时间纬度
     */
    private void handleDataRecordGroupByColumnSdate(SelectColumnNameAndAlias target,
            List<String> dataRecordGroupColumns, TimeDimension timeDimension) {
        int indexOfSdateGroupBy = dataRecordGroupColumns.indexOf(target.columnName);
        if (indexOfSdateGroupBy >= 0) {
            dataRecordGroupColumns.set(indexOfSdateGroupBy, dataRecordColumnNameOfSdate(timeDimension));
        }
    }

    /**
     * 不同纬度下的sdate格式化
     *
     * @param timeDimension 时间纬度
     * @return 格式化内容
     */
    private String dataRecordColumnNameOfSdate(TimeDimension timeDimension) {
        String result;
        switch (timeDimension) {
            case Day:
                result = "sdate";
                break;
            case Week:
                result = "concat(year(sdate),'-',weekofyear(sdate))";
                break;
            case Month:
                result = "concat(year(sdate),'-',month(sdate))";
                break;
            case Year:
                result = "year(sdate)";
                break;
            default:
                result = "year(sdate)";
        }
        return result;
    }

    private void assembleAffiliateCsGeneralData(ReportDataHelper helper, AffiliateReportParamDTO param,
            ReportParam reportParam, TimeDimension timeDimension, MergeDayStatCsGeneralDO mergeRealTimeByRMB,
            MergeDayStatCsGeneralDO mergeRealTimeByDollar, TimeDimension originTimeDimension,
            List<SelectColumnNameAndAlias> csGeneralSelectColumns) {
        final List<OfferReportField> options = param.getOfferReportFieldList();

        List<OrderByColumn> orderByColumns = options.stream()
                .map(AffiliateMergeColumnAndAlias::orderBy)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (orderByColumns == null) {
            orderByColumns = new ArrayList<>();
        }
        orderByColumns.add(new OrderByColumn("id", "desc"));

        final List<String> groupByColumns = options.stream()
                .map(AffiliateMergeColumnAndAlias::groupBy)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());

        final List<MergeDayStatCsGeneralDO> mergeDayStatData = dayStatCsGeneralMapper.mergeDataFromDayStatCsGeneral(
                reportParam, csGeneralSelectColumns, groupByColumns, orderByColumns,
                timeZoneSubService.prefixTableNameForTimeZone(param.getTimeZone()),
                timeDimension.key);
        logger.debug("{}", mergeDayStatData);

        helper.addAllDayStatCsGeneralMergeData(mergeDayStatData);

        final JsonMapper jsonMapper = JsonMapper.nonEmptyMapper();
        logger.info("csGeneral reportParam:{},selectParam:{}, groupByColumns:{}, orderByColumns:{} mapDataSize:{}, timeDimension:{}",
                jsonMapper.toJson(reportParam), jsonMapper.toJson(csGeneralSelectColumns),
                jsonMapper.toJson(groupByColumns), jsonMapper.toJson(orderByColumns),
                mergeDayStatData.size(), timeDimension.key);

        /* 2. 添加实时数据 */
        boolean hourOrDay = originTimeDimension == TimeDimension.Hour || originTimeDimension == TimeDimension.Day;
        boolean includeNow = isIncludeNow(param);
        boolean affiliateGroupMatch = affiliateGroupMatch(options);
        logger.info("inCloudeNow:{}, affilateGroupMatch:{}", includeNow, affiliateGroupMatch);
        if (hourOrDay && includeNow && affiliateGroupMatch) {
            List<RealTimeGroupByField> fields = options.stream()
                    .map(RealTimeGroupByField::of)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
            addRealTimeData(param.getTimeZone(), timeDimension,
                    fields, helper, mergeRealTimeByRMB, mergeRealTimeByDollar);
        }
    }

    private void handleCsGeneralDataSelectColumnSdate(SelectColumnNameAndAlias target,
            List<SelectColumnNameAndAlias> csGeneralSelectColumns, TimeDimension timeDimension) {
        if (CollectionUtils.isNotEmpty(csGeneralSelectColumns)) {
            SelectColumnNameAndAlias selectColumnNameAndAlias = csGeneralSelectColumns.stream()
                    .filter(item -> item.columnName.equalsIgnoreCase(target.columnName))
                    .findFirst()
                    .orElse(null);
            if (selectColumnNameAndAlias != null) {
                // 这里如果不创建一个新的，会将类变量的值变更掉，导致前面的操作影响后面的操作
                selectColumnNameAndAlias.setColumnName(csGeneralDataColumnNameOfSdate(timeDimension));
            }
        }
    }

    private String csGeneralDataColumnNameOfSdate(TimeDimension timeDimension) {
        String result;
        switch (timeDimension) {
            case Hour:
                result = "concat(left(sdate, 4), '-', substring(sdate,5,2), '-',substring(sdate,7,2), ' ', right(sdate,2)) ";
                break;
            case Day:
                result = "concat(left(sdate, 4), '-', substring(sdate,5,2), '-',substring(sdate,7,2))";
                break;
            case Week:
                result = "concat(year(sdate), '-', week(sdate))";
                break;
            case Month:
                result = "concat(year(sdate), '-', month(sdate))";
                break;
            case Year:
                result = "year(sdate)";
                break;
            default:
                result = "sdate";
        }
        return result;
    }

    @SuppressWarnings("all")
    private boolean affiliateGroupMatch(List<OfferReportField> options) {
        for (OfferReportField offerReportField : options) {
            if (AFFILIATE_NOT_MATCH_GROUPS.contains(offerReportField.getField())) {
                return false;
            }
        }
        return true;
    }


    private ReportParam validAndConstructReportParam(ReportBaseParamDTO param) {
        /* startDate,endDate 已经是指定时区的时间 可以直接用于时区表中sdate的过滤 */
        final Integer startDate = param.getStartDate();
        final Integer endDate = param.getEndDate();
        final Integer timeZone = param.getTimeZone();
        final TimeDimension timeDimension = param.getTimeDimension();
        Assert.notNull(startDate, "StartDate not exists");
        Assert.notNull(endDate, "EndDate not exists");
        Assert.notNull(timeZone, "TimeZone not exists");
        Assert.notNull(timeDimension, "timeDimension can not be null");
        Assert.notNull(param.getReportType(), "offerType can not be null");
        Assert.isFalse(startDate > endDate, "Start date can bot be greater than end date");
        final ReportParam reportParam = new ReportParam();
        reportParam.setStartDate(timeDimensionWithSdate(String.valueOf(startDate), timeZone, timeDimension).getSdate());
        reportParam.setEndDate(timeDimensionWithSdate(String.valueOf(endDate), timeZone, timeDimension).getSdate());
        reportParam.setCompanyId(param.getCompanyId());
        reportParam.setReportType(param.getReportType().getKey());
        reportParam.setOfferIds(Optional.ofNullable(param.getOfferIds())
                .map(item -> item.stream().map(String::valueOf).collect(Collectors.toList())).orElse(null));
        reportParam.setFromLine((param.getPage() - 1) * param.getRows());
        reportParam.setRows(param.getAllRows());
        return reportParam;
    }

    /**
     * 实时数据 再次合并实时数据
     */
    private void addRealTimeData(Integer timeZone, TimeDimension timeDimension,
            List<RealTimeGroupByField> realTimeGroupByFields, ReportDataHelper helper,
            MergeDayStatCsGeneralDO mergeRealTimeByRMB, MergeDayStatCsGeneralDO mergeRealTimeByDoller) {

        final JsonMapper jsonMapper = JsonMapper.nonEmptyMapper();
        AllRedisRealTimeData allRedisRealTimeData = realTimeDataDao.allRedisRealTimeData();
        helper.forEach(offerReportData -> {
            RealTimeDataSearchParam searchParam = new RealTimeDataSearchParam();
            final Instant[] instant = toUTCInstantBeginAndEnd(offerReportData.getSdates(), timeZone, timeDimension);
            if (instant != null) {
                searchParam.setOfferId(offerReportData.getOfferId());
                searchParam.setAffiliateId(offerReportData.getAffiliateId());
                searchParam.setAdvertiserId(offerReportData.getAdvertiserId());
                /* 检索并聚合所有实时数据 */
                List<RealTimeData> realTimeDatas = realTimeDataDao.mergeRealTimeData(
                        allRedisRealTimeData, realTimeGroupByFields, searchParam, instant[0], instant[1]);
                logger.debug("RealTimeData, size:{}, data:{}", realTimeDatas.size(), jsonMapper.toJson(realTimeDatas));
                /* 合并实时数据到统计数据 */
                if (mergeRealTimeByRMB.getSymbol().equals(offerReportData.getSymbol())) {
                    realTimeDatas.forEach(realTimeData -> realTimeData.addToOfferReportData(offerReportData, mergeRealTimeByRMB));
                } else if (mergeRealTimeByDoller.getSymbol().equals(offerReportData.getSymbol())) {
                    realTimeDatas.forEach(realTimeData -> realTimeData.addToOfferReportData(offerReportData, mergeRealTimeByDoller));
                } else {
                    logger.info("This realTimeData has not Symbol, data:{}", jsonMapper.toJson(realTimeDatas));
                }
            }
        });
    }

    @Override
    public List<TopOfferConversionResultDTO> topOfferConversions(TopOfferConversionParamDTO param) {
        Assert.notNull(param.getAdvertiserId(), "Invalid advertiser id");
        Assert.notNull(param.getBegin(), "Invalid begin date");
        Assert.notNull(param.getEnd(), "Invalid end date");

        List<TopOfferConversionDO> dos = dayStatCsAdvOfferMapper
                .topOfferConversions(param, DateUtils.dateToInt(param.getBegin()), DateUtils.dateToInt(param.getEnd()), timeZoneSubService.prefixTableNameForTimeZone(param.getTimeZone()));
        return CollectionUtils.isEmpty(dos) ?
               null :
               dos.stream().filter(Objects::nonNull).map(TopOfferConversionDO::toDto).collect(Collectors.toList());
    }

    @Override
    public PageResult<BillInOutReportDTO> findCsGeneralByParam(BillInOutReportParam param) {
        List<BillInOutReportDTO> billInOutReportDTOS = dayStatCsGeneralMapper.findCsGeneralByParam(param, timeZoneSubService.prefixTableNameForTimeZone(param.getZone()));
        Integer total = dayStatCsGeneralMapper.findCsGeneralByParamCount(param, timeZoneSubService.prefixTableNameForTimeZone(param.getZone()));
        PageResult<BillInOutReportDTO> pageResult = new PageResult<>();
        pageResult.setTotal(total);
        pageResult.setRows(billInOutReportDTOS);
        return pageResult;
    }

    private static final List<AdvertiserOfferReportGroupByField> ADVERTISER_NOT_MATCH_GROUPS = Lists
            .newArrayList(AdvertiserOfferReportGroupByField.AffiliateSource,
                    AdvertiserOfferReportGroupByField.Sub1,
                    AdvertiserOfferReportGroupByField.Sub2,
                    AdvertiserOfferReportGroupByField.Sub3,
                    AdvertiserOfferReportGroupByField.Sub4,
                    AdvertiserOfferReportGroupByField.Sub5
            );

    private static final List<AffiliateOfferReportGroupByField> AFFILIATE_NOT_MATCH_GROUPS = Lists
            .newArrayList(AffiliateOfferReportGroupByField.AffiliateSource, AffiliateOfferReportGroupByField.Sub1,
                    AffiliateOfferReportGroupByField.Sub2, AffiliateOfferReportGroupByField.Sub3,
                    AffiliateOfferReportGroupByField.Sub4, AffiliateOfferReportGroupByField.Sub5);


    private static boolean isIncludeNow(ReportBaseParamDTO param) {
        ZoneId zoneId = DateUtils.zoneId(param.getTimeZone());
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMddHHmmss").withZone(zoneId);
        TemporalAccessor sdateS = formatter.parse(String.valueOf(param.getStartDate()) + "0000");
        TemporalAccessor sdateE = formatter.parse(String.valueOf(param.getEndDate()) + "5959");
        ZonedDateTime utc = ZonedDateTime.now(zoneId);
        return
                (sdateS.get(ChronoField.YEAR) <= utc.getYear() && sdateS.get(ChronoField.DAY_OF_YEAR) <= utc.getDayOfYear())
                        && (sdateE.get(ChronoField.YEAR) >= utc.getYear() && sdateE.get(ChronoField.DAY_OF_YEAR) >= utc
                        .getDayOfYear());
    }

    private Instant[] toUTCInstantBeginAndEnd(String sdate, Integer timeZone, TimeDimension timeDimension) {
        if (timeDimension == TimeDimension.Day || timeDimension == TimeDimension.Hour) {
            if (StringUtils.isNotBlank(sdate)) {
                sdate = sdate.replaceAll("-", "").replaceAll(" ", "");
                boolean day = timeDimension == TimeDimension.Day;
                sdate = day ? sdate.concat("000000") : sdate.concat("0000");
                Date date = DateUtils.parse(sdate, "yyyMMddHHmmss", timeZone);
                /* 0时区时间 */
                Instant instant1 = date.toInstant();
                Instant instant2 = date.toInstant().plus(1, day ? ChronoUnit.DAYS : ChronoUnit.HOURS).minusSeconds(1);
                return new Instant[]{instant1, instant2};
            } else {
                /* 总合计记录 sdate为空 */
                Instant end = Instant.now();
                return new Instant[]{end.minus(1, ChronoUnit.DAYS), end};
            }
        } else {
            return null;
        }


    }

    private static class ReportDataHelper {
        private List<OfferReportDataDTO> data = new ArrayList<>();
        private Map<ReportDataKey, OfferReportDataDTO> map = new HashMap<>();

        List<OfferReportDataDTO> data() {
            return data;
        }

        void forEach(Consumer<OfferReportDataDTO> consumer) {
            data.forEach(consumer);
        }

        ReportDataHelper addAllDayStatCsGeneralMergeData(List<MergeDayStatCsGeneralDO> data) {
            if (CollectionUtils.isNotEmpty(data)) {
                data.forEach(this::add);
            }
            return this;
        }

        private void add(MergeDayStatCsGeneralDO item) {
            ReportDataKey key = new ReportDataKey(item);
            map.compute(key, (k, v) -> {
                if (v == null) {
                    OfferReportDataDTO offerReportData = item.toOfferReportDataDTO();
                    this.data.add(offerReportData);
                    return offerReportData;
                } else {
                    merge(item, v);
                    return v;
                }
            });
        }

        ReportDataHelper addAllMergeDataRecord(List<MergeDataRecordDO> data) {
            if (CollectionUtils.isNotEmpty(data)) {
                data.forEach(this::add);
            }
            return this;
        }

        private void add(MergeDataRecordDO item) {
            ReportDataKey key = new ReportDataKey(item);
            map.compute(key, (k, v) -> {
                if (v == null) {
                    OfferReportDataDTO offerReportData = ReportDataHelper.toAdvertReportData(item);
                    this.data.add(offerReportData);
                    return offerReportData;
                } else {
                    merge(item, v);
                    return v;
                }
            });
        }

        @SuppressWarnings("all")
        static void merge(MergeDataRecordDO source, OfferReportDataDTO target) {
            target.setClicks(add(target.getClicks(), source.getClicks()));
            target.setUniqueClicks(add(target.getUniqueClicks(), source.getUniqueClicks()));
            target.setGrossConversions(add(target.getGrossConversions(), source.getGrossConversions()));
            target.setScrubbedConversions(add(target.getScrubbedConversions(), source.getScrubbedConversions()));
            target.setConversions(add(target.getConversions(), source.getConversions()));
            target.setRevenue(add(target.getRevenue(), source.getRevenue()));
            target.setCost(add(target.getCost(), source.getCost()));
        }

        @SuppressWarnings("all")
        static void merge(MergeDayStatCsGeneralDO source, OfferReportDataDTO target) {
            target.setClicks(add(target.getClicks(), source.getClicks()));
            target.setUniqueClicks(add(target.getUniqueClicks(), source.getUniqueClicks()));
            target.setGrossConversions(add(target.getGrossConversions(), source.getGrossConversions()));
            target.setScrubbedConversions(add(target.getScrubbedConversions(), source.getScrubbedConversions()));
            target.setConversions(add(target.getConversions(), source.getConversions()));
            target.setRevenue(add(target.getRevenue(), source.getRevenue()));
            target.setCost(add(target.getCost(), source.getCost()));
        }

        static int add(Integer i1, Integer i2) {
            return com.codrim.common.utils.math.NumberUtils.toInt(i1) + com.codrim.common.utils.math.NumberUtils
                    .toInt(i2);
        }

        static long add(Long t1, Long t2) {
            return Optional.ofNullable(t1).orElse(0L)
                    + Optional.ofNullable(t2).orElse(0L);
        }

        static float add(Float f, Float f2) {
            f = f == null ? 0f : f;
            f2 = f2 == null ? 0f : f2;
            return f + f2;
        }

        static OfferReportDataDTO toAdvertReportData(MergeDataRecordDO data) {
            final OfferReportDataDTO result = new OfferReportDataDTO();
            result.setSdates(data.getSdates());
            String advertiserId = data.getAdvertiserId();
            String affiliateId = data.getAffiliateId();
            String offerId = data.getOfferId();
            result.setSymbol(data.getSymbol());
            result.setCustomerId(data.getCustomerId());
            result.setAdvertiserId(StringUtils.isNotBlank(advertiserId) ? Integer.valueOf(advertiserId) : null);
            result.setAffiliateId(StringUtils.isNotBlank(affiliateId) ? Integer.valueOf(affiliateId) : null);
            result.setOfferId(StringUtils.isNotBlank(offerId) ? Integer.valueOf(offerId) : null);
            result.setAdvAffiliateManagerId(data.getAdvAffiliateManagerId());
            result.setAdvAdvertiserManagerId(data.getAdvAdvertiserManagerId());

            result.setClicks(data.getClicks());
            result.setUniqueClicks(data.getUniqueClicks());
            result.setGrossConversions(data.getGrossConversions());
            result.setConversions(data.getConversions());
            result.setScrubbedConversions(data.getScrubbedConversions());
            result.setRevenue(data.getRevenue());
            result.setCost(data.getCost());
            result.setAccessType(data.getAccessType());

            return result;
        }
    }


    @Data
    private static class ReportDataKey {
        private String sdates;
        private Integer advertiserId;
        private Integer affiliateId;
        private Integer customerId;
        private Integer advAdvertiserManagerId;
        private Integer advAffiliateManagerId;
        private Integer affAdvertiserManagerId;

        private Integer offerId;
        private String symbol;

        private String sourceId;
        private String sub1;
        private String sub2;
        private String sub3;
        private String sub4;
        private String sub5;

        ReportDataKey(MergeDayStatCsGeneralDO data) {
            this.sdates = data.getSdates();
            this.advertiserId = data.getAdvertiserId();
            this.affiliateId = data.getAffiliateId();
            this.offerId = data.getOfferId();
            this.customerId = data.getCustomerId();
            this.advAdvertiserManagerId = data.getAdvAdvertiserManagerId();
            this.advAffiliateManagerId = data.getAdvAffiliateManagerId();
            this.affAdvertiserManagerId = data.getAffAdvertiserManagerId();
            this.setSymbol(data.getSymbol());

            this.sourceId = data.getSourceId();
            this.sub1 = data.getSub1();
            this.sub2 = data.getSub2();
            this.sub3 = data.getSub3();
            this.sub4 = data.getSub4();
            this.sub5 = data.getSub5();
        }

        ReportDataKey(MergeDataRecordDO data) {
            this.symbol = data.getSymbol();
            this.sdates = data.getSdates();
            this.customerId = data.getCustomerId();
            this.advAffiliateManagerId = data.getAdvAffiliateManagerId();
            this.advAdvertiserManagerId = data.getAdvAdvertiserManagerId();
            this.affAdvertiserManagerId = data.getAffAdvertiserManagerId();
            String advertiserId = data.getAdvertiserId();
            String affiliateId = data.getAffiliateId();
            String offerId = data.getOfferId();
            this.advertiserId = StringUtils.isNotBlank(advertiserId) ? Integer.valueOf(advertiserId) : null;
            this.affiliateId = StringUtils.isNotBlank(affiliateId) ? Integer.valueOf(affiliateId) : null;
            this.offerId = StringUtils.isNotBlank(offerId) ? Integer.valueOf(offerId) : null;
        }
    }
}
